package com.hudsonmq.spring.handler.process;

import com.hudson.codec.domain.MessageExt;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.util.CollectionUtil;
import com.hudsonmq.spring.domain.consume.PushRequest;
import com.hudsonmq.spring.handler.Process;
import com.hudsonmq.spring.impl.PushMessageServiceOneThread;
import com.hudsonmq.spring.impl.consume.management.ConsumerHandlerManagement;
import com.hudsonmq.spring.impl.consume.management.PushMessageMonitor;
import com.hudsonmq.spring.utils.SpringContextUtil;
import com.hudsonmq.spring.utils.ThreadPoolHelper;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Author: 渔小鲤
 * @DateTime: 2025/7/25 22:57
 **/
@Component
@ConditionalOnProperty(prefix = "hudson",name = "consume-group")
public class PushMessageHandler implements Process {
    private static final Logger logger = LoggerFactory.getLogger(PushMessageHandler.class);

    private ConsumerHandlerManagement consumerHandlerManagement;


    @Override
    public String getType() {
        return MessageTypeEnum.PUSH_MESSAGE_RESP.getType();
    }

    @Override
    public RemotingCommand process(ChannelHandlerContext channelHandlerContext, RemotingCommand remotingCommand) {
        if(consumerHandlerManagement == null){
            consumerHandlerManagement = SpringContextUtil.getBean(ConsumerHandlerManagement.class);
        }
        List<MessageExt> messageExtList = MessageExt.parse2List(remotingCommand.getBody());
        MessageExt messageInfo = messageExtList.get(0);
        if (CollectionUtil.noEmpty(messageInfo.getBody())) {
            consumerHandlerManagement.exec(() -> consumerHandlerManagement.onMessage(channelHandlerContext.channel(), messageExtList));
            PushMessageMonitor.incrementQueueSize(messageInfo.getTopic(), messageInfo.getQueueId(), messageExtList.size());
        }


        if (PushMessageMonitor.isMax(messageInfo.getTopic(), messageInfo.getQueueId())) {
            PushMessageMonitor.decrementPullMessageSize(messageInfo.getTopic(), messageInfo.getQueueId());
            ThreadPoolHelper.DEFAULT_SCHEDULED_POOL.schedule(() -> {
                Integer pullMessageSize = PushMessageMonitor.getPullMessageSize(messageInfo.getTopic(), messageInfo.getQueueId());
                PushMessageServiceOneThread.pushRequest(new PushRequest(channelHandlerContext.channel(), messageInfo.getTopic(), messageInfo.getQueueId(), pullMessageSize));
            }, 2, TimeUnit.SECONDS);
        } else if (CollectionUtil.isEmpty(messageInfo.getBody())) {
            ThreadPoolHelper.DEFAULT_SCHEDULED_POOL.schedule(() -> {
                Integer pullMessageSize = PushMessageMonitor.getPullMessageSize(messageInfo.getTopic(), messageInfo.getQueueId());
                PushMessageServiceOneThread.pushRequest(new PushRequest(channelHandlerContext.channel(), messageInfo.getTopic(), messageInfo.getQueueId(), pullMessageSize));
            }, 1, TimeUnit.SECONDS);
        } else {
            Integer pullMessageSize = PushMessageMonitor.getPullMessageSize(messageInfo.getTopic(), messageInfo.getQueueId());
            PushMessageServiceOneThread.pushRequest(new PushRequest(channelHandlerContext.channel(), messageInfo.getTopic(), messageInfo.getQueueId(), pullMessageSize));
            logger.debug("客户端收到服务端消息: " + remotingCommand);
        }

        return null;
    }
}
